package chenxu.test.jms;

import java.io.IOException;
import java.io.InputStream;

import javax.jms.JMSException;
import javax.jms.Message;

import org.apache.activemq.blob.BlobDownloader;
import org.apache.activemq.blob.BlobTransferPolicy;
import org.apache.activemq.command.ActiveMQBlobMessage;
import org.apache.activemq.util.Callback;
import org.apache.commons.io.IOUtils;

/**
 *	
 *
 */
public class ActiveMQBlobMessageUtil {

	/**
	 * 除了把FTP上的文件转换成字符串外，还要给每一个 Message 设置一个
	 * AcknowledgeCallback，当消息被确认后此此消息对应的FTP上的文件
	 * 会被删除，否则文件保留在FTP上，供消息被重放的时候再次读取（activemq
	 * 重放消息的时候不会重新上传文件）
	 * @param message
	 * @return
	 * @throws JMSException
	 */
	public static String toString(Message message) throws JMSException {

		if(message instanceof ActiveMQBlobMessage){
			ActiveMQBlobMessage blobMsg = (ActiveMQBlobMessage) message;
			blobMsg.setAcknowledgeCallback(new DelFileCallback(blobMsg));
			InputStream inputStream = null;
			BlobTransferPolicy transferPolicy = new BlobTransferPolicy();
			BlobDownloader blobDownloader = new BlobDownloader(transferPolicy );
			try {
				transferPolicy.setUploadUrl(blobMsg.getURL().toString());
				blobMsg.setBlobDownloader(blobDownloader );
				inputStream = blobMsg.getInputStream();
				String values = IOUtils.toString(inputStream);
				return values;
			} catch (IOException e) {
				throw new JMSException(e.getMessage());
			} finally {
				try {
					if(null!=inputStream)
						inputStream.close();
				} catch (IOException e) {
					throw new JMSException("close "+blobMsg.getMessageId()+"'s inputstream occor error");
				}
			}
		} else {
			return null;
		}
	}
	
	static class DelFileCallback implements Callback {
		
		ActiveMQBlobMessage msg;

		public DelFileCallback(ActiveMQBlobMessage msg) {
			this.msg = msg;
		}
		
		@Override
		public void execute() throws Exception {
			msg.deleteFile();
		}
		
	}
		
}
